package code.nest.util.kafka.impl;

import cn.hutool.json.JSONUtil;
import code.nest.util.kafka.KafkaProducerService;
import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.clients.producer.internals.DefaultPartitioner;

import java.util.Properties;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;

/**
 * @author ：Hyman
 * @date ：Created in 2021/6/22 16:16
 * @description：
 * @modified By：
 * @version: $
 */
public class KafkaProceducerServiceImpl<T> implements KafkaProducerService<T> {



    public KafkaProceducerServiceImpl() {

    }

    private <V> KafkaProducer createKafkaProducer(String keySerializer, String valueSerializer) {
        Properties properties = new Properties();
        /* kafka的链接信息*/
        properties.put("bootstrap.servers", "172.17.6.85:9092");
        /* 键的序列化器 */
        properties.put("key.serializer", keySerializer);
        /* 值得序列化器 */
        properties.put("value.serializer", valueSerializer);
        /* ack=0，生产者在写入消息到代理后不会等待服务器的响应，吞吐量最高
         *  ack=1，只要集群的首领节点收到消息，生成者就会收到一个来自服务器的成功响应，这个时候的吞吐量取决于是采用异步发送还是同步发送
         *  ack=all，当所有同步副本全部接受到消息时，生产者才会收到一个来自服务器的成功响应，这种模式最安全。
         * */
        properties.put("acks", "all");
        /* 设置生成者内存缓冲区的大小，生成者的消息发送到服务器之前先在缓冲区缓冲
         *  单位byte */
        properties.put("buffer.memory", "16384");
        /* 生成者的内存缓冲区大小不足时，send方法阻塞的毫秒数，过了这个毫秒数之后send会抛出异常。*/
        properties.put("max.block.ms", "2000");
        /* 压缩类型，默认不开启压缩。使用压缩可以降低网络传输开销和存储开销，但是容易成为性能瓶颈。
         * compression.type=snappy，占用较少的CPU，提供较好的性能和相当可观的压缩比。
         * compression.type=gzip，占用较多的CPU，提高更高的压缩比，适合低网络带宽。
         * compression.type=lz4
         * */
        properties.put("compression.type", "snappy");
        /* 生成者从服务器收到错误之后的重试次数，每次重试之间间隔100ms，间隔
         * 可以通过参数retry.backoff.ms设置*/
        properties.put("reties", "3");
        /* 设置重试间隔时间 */
        properties.put("retry.backoff.ms", "50");
        /* 当有多个消息被发送到同一个分区时，生产者会把他们放到同一个批次，
         * 该参数指定一个批次可以使用的内存大小，单位byte*/
        properties.put("batch.size", "1024");
        /* 生产者生产的消息批次填满或者达到linger.ms上限时会把这批次发送出去
         * 默认是0，设置了会增加延迟，但是会增加吞吐量*/
        properties.put("linger.ms", "1000");
        /* 设置客户端的id */
        properties.put("client.id", "myclient");
        /* 指定生产者在接受到服务器响应之前可以发送多少个消息，值越高，需要越多内存，
         * 但是会提高吞吐量，设置为1可以保证消息按照发送顺序写入到服务器*/
        properties.put("max.in.flight.requests.per.connection", "1");
        /* 指定生产者发送数据时等待服务器返回响应的时间 */
        properties.put("request.timeout.ms", "200");
        /* 指定生产者向服务器获取元数据（比如目标分区首领是谁）的超时时间 */
        properties.put("metadata.fetch.timeout.ms", "200");
        /* 指定生产者同步副本获取响应的超时时间 */
        properties.put("timeout.ms", "200");
        /* 指定单个请求中发送消息的最大值 */
        properties.put("max.request.size", "1024");
        /* 指定TCP Socket发送缓冲区的大小，被设置成-1后，使用操作系统的默认值
         *  receive.buffer.bytes是指定接收缓冲区的大小，如果生产者和消费者处于
         * 不同的数据中心，可以适当增大这些值，因为跨数据中心网络一般都有比较高的
         * 延迟和比较低的带宽*/
        properties.put("send.buffer.bytes", "256");
        /* 默认分区器 */
        properties.put("partitioner.class", "org.apache.kafka.clients.producer.internals.DefaultPartitioner");
        /* 自定义的分区器 */
        //properties.put("partitioner.class", "code.nest.partitioner.MyPartitioner");
        KafkaProducer kafkaProducer = new KafkaProducer<String, V>(properties);
        return kafkaProducer;
    }

    @Override
    public void sendFireAndForgetMsg(String topic, String key, T t) {
        ProducerRecord<String, String> record = new ProducerRecord<>(topic, key, JSONUtil.toJsonStr(t));
        try {
            /* 采用StringSerializer序列化器 */
            KafkaProducer kafkaProducer = this.<String>createKafkaProducer(
                    "org.apache.kafka.common.serialization.StringSerializer",
                    "org.apache.kafka.common.serialization.StringSerializer"
            );
            kafkaProducer.send(record);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }


    @Override
    public void sendFutureMsg(String topic, String key, T t) {
        try {
            /* 采用自定义的序列化器LogSerializer */
            KafkaProducer kafkaProducer = this.<T>createKafkaProducer(
                    "org.apache.kafka.common.serialization.StringSerializer",
                    "code.nest.serializer.LogSerializer"
            );
            ProducerRecord<String, T> record = new ProducerRecord<>(topic, key, t);
            Future<RecordMetadata> future = kafkaProducer.send(record);
            RecordMetadata recordMetadata = future.get();
            System.out.println(recordMetadata);
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }

    }

    @Override
    public void sendAsynMsg(String topic, String key, T t) {
        KafkaProducer kafkaProducer = createKafkaProducer(
                "org.apache.kafka.common.serialization.StringSerializer",
                "code.nest.serializer.ProtobufSerializer"
        );
        ProducerRecord<String, T> record = new ProducerRecord<>(topic, key, t);
        kafkaProducer.send(record, new ProducerCallBack());
    }

    /**
     * create by: Hyman
     * description: 这是一个回调类，在异步发送方式中，调用该类对象的onCompletion方法。
     * create time: 2021/6/22
     */
    public static class ProducerCallBack implements Callback {


        @Override
        public void onCompletion(RecordMetadata recordMetadata, Exception e) {
            if (e != null) {
                e.printStackTrace();
            } else {
                System.out.println("异步消息发送成功！");
            }
        }
    }
}
